package com.zj.dmsproxy.utils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @author 19242
 */
public class AsyncUtil<T>{
    /**
     * 线程池
     */
    private ThreadPoolExecutor executor;
    /**
     * 要并发的原数据
     */
    private List<T> dataList;

    private AsyncUtil(List<T> dataList, int corePoolSize, int maximumPoolSize) {
        this.dataList = dataList;
        // 初始化线程池
        initializeExecutor(corePoolSize, maximumPoolSize);
    }

    private AsyncUtil(List<T> dataList, ThreadPoolExecutor executor){
        this.dataList = dataList;
        this.executor = executor;
    }

    /**
     * 放入要并发的数据
     * @param dataList
     * @param <T>
     * @return
     */
    public static <T> AsyncUtil<T> of(List<T> dataList){
        return new AsyncUtil<>(dataList, 50, 200);
    }
    /**
     * 放入要并发的数据和自定义的线程线程数
     * @param dataList
     * @param <T>
     * @return
     */
    public static <T> AsyncUtil<T> of(List<T> dataList, int corePoolSize, int maximumPoolSize){
        return new AsyncUtil<>(dataList, corePoolSize, maximumPoolSize);
    }

    /**
     * 放入要并发的数据和自定义的线程
     * @param dataList
     * @param executor
     * @param <T>
     * @return
     */
    public static <T> AsyncUtil<T> of(List<T> dataList, ThreadPoolExecutor executor){
        return new AsyncUtil<>(dataList, executor);
    }
    /**
     * 创建一个线程异步运行
     * @return
     */
    public static void runAsync(Runnable runnable){
        CompletableFuture.runAsync(runnable);
    }

    /**
     * 将数组拆成每size一组
     * @param list 要拆分的数组
     * @param size 每组多少个
     * @param <T>
     * @return
     */
    public static <T> List<List<T>> splitListBySize(List<T> list, int size){
        if (Objects.isNull(list) || list.isEmpty() || size <= 0){
            return Collections.emptyList();
        }
        List<List<T>> resultList = new ArrayList<>();
        int i = size;
        for (; i < list.size(); i += size) {
            List<T> children = list.subList(i - size, i);
            resultList.add(children);
        }
        List<T> children = list.subList(i - size, list.size());
        resultList.add(children);
        return resultList;
    }

    /**
     * 将数组拆成n组
     * @param list 要拆分的数组
     * @param n 分成几组 如果大于list.size，分成list.size组
     * @param <T>
     * @return
     */
    public static <T> List<List<T>> splitListToN(List<T> list, int n){
        if (Objects.isNull(list) || list.isEmpty() || n <= 0){
            return Collections.emptyList();
        }
        // 每组平均多少个
        int size = list.size() / n;
        // n>list.size()时
        if (size == 0){
            return splitListBySize(list, 1);
        }
        List<List<T>> resultList = new ArrayList<>();
        // 平均后余出的数量
        int over = list.size() % n;
        int i = size;
        // 余出的数据均分给前over个数组
        int flag;
        for (; i < list.size(); i += size + flag) {
            flag = 0;
            if (over > 0){
                over--;
                flag = 1;
            }
            List<T> children = list.subList(i - size, i + flag);
            resultList.add(children);
        }
        List<T> children = list.subList(i - size, list.size());
        resultList.add(children);
        return resultList;
    }

    /**
     * 没有返回值的并发操作 异步
     * @param consumer
     */
    public void runAsync(Consumer<T> consumer){
        run(consumer, false);
    }

    /**
     * 有返回值的并发操作 异步
     * @param func
     * @param <U>
     * @return
     */
    public <U> List<U> supplyAsync(Function<T, U> func){
        return supply(func, false);
    }

    /**
     * 没有返回值的并发操作 同步
     * @param consumer
     */
    public void runAsyncJoin(Consumer<T> consumer){
        run(consumer, true);
    }

    /**
     * 有返回值的并发操作 同步
     * @param func
     * @param <U>
     * @return
     */
    public <U> List<U> supplyAsyncJoin(Function<T, U> func){
        return supply(func, true);
    }

    /**
     * 没有返回值的并发操作
     * @param consumer
     * @param isJoin
     */
    private void run(Consumer<T> consumer, boolean isJoin){

        CompletableFuture[] completableFutures = new CompletableFuture[dataList.size()];
        for (int i = 0; i < dataList.size(); i++) {
            T s = dataList.get(i);
            completableFutures[i] = CompletableFuture.runAsync(() -> consumer.accept(s), executor);
        }
        CompletableFuture<Void> allOf = CompletableFuture.allOf(completableFutures).exceptionally(throwable -> {
            System.out.println(throwable.getMessage());
            throw new RuntimeException(throwable);
        });
        if (isJoin){
            allOf.join();
        }
        executor.shutdown();
    }

    /**
     * 有返回值的并发操作
     * @param func
     * @param <U>
     * @return
     */
    private <U> List<U> supply(Function<T, U> func, boolean isJoin){

        CompletableFuture<Void>[] completableFutures = new CompletableFuture[dataList.size()];
        // 用线程安全的list收集结果
        List<U> resultList = Collections.synchronizedList(new ArrayList<>());
        for (int i = 0; i < dataList.size(); i++) {
            T s = dataList.get(i);
            // 不用supplyAsync方法 收集返回结果。因为还要遍历completableFutures
            completableFutures[i] = CompletableFuture.runAsync(() -> resultList.add(func.apply(s)), executor);

        }
        CompletableFuture<Void> allOf = CompletableFuture.allOf(completableFutures).exceptionally(throwable -> {
            System.out.println(throwable.getMessage());
            throw new RuntimeException(throwable);
        });
        if (isJoin){
            allOf.join();
        }
        executor.shutdown();
        return resultList;
    }

    /**
     * 初始化线程池
     * @param corePoolSize
     * @param maximumPoolSize
     */
    private void initializeExecutor(int corePoolSize, int maximumPoolSize) {
        executor = new ThreadPoolExecutor(corePoolSize,
                maximumPoolSize,
                6,
                TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(dataList.size() > maximumPoolSize ? dataList.size() - maximumPoolSize : corePoolSize),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

}
